package com.mzj.eagle.jobs.config.emq;

import java.util.Arrays;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mzj.eagle.jobs.config.sys.SpringUtils;


public class PushCallback implements MqttCallback {
	
	private static final Logger LOG = LoggerFactory.getLogger(PushCallback.class);
	
	private static final String VERSION_KEY = "version";
	private static final String DATA_KEY = "data";

	private String routingKey;
	
	private ObjectMapper mapper = new ObjectMapper();
	
	public PushCallback(String routingKey) {
		super();
		this.routingKey = routingKey;
	}

	public PushCallback() {
		super();
	}

	public void connectionLost(Throwable cause) {
		// 连接丢失后，一般在这里面进行重连
		LOG.error("EMQ连接失败：{}",cause.getMessage());
	}

	public void deliveryComplete(IMqttDeliveryToken token) {
		// publish后会执行到这里
		LOG.info("EMQ push message: {},{}",Arrays.toString(token.getTopics()),token.isComplete());
	}

	public void messageArrived(String topic, MqttMessage message){
		// subscribe后得到的消息会执行到这里面
		LOG.info("接收消息主题: {}",topic);
		LOG.info("接收消息Qos: {}",message.getQos());
		String messageSt = new String(message.getPayload());
		LOG.info("接收消息内容: {}",messageSt);
		JsonNode jsonNode = isJson(mapper,messageSt);
		if(ObjectUtils.isEmpty(jsonNode)) {
			return;
		}
		
		String version = jsonNode.get(VERSION_KEY).asText();
		String data = jsonNode.get(DATA_KEY).toString();
		RabbitTemplate mqAgency = SpringUtils.getBean(RabbitTemplate.class);
		mqAgency.convertAndSend(routingKey, data);
	}

	/**
	 * 判断消息是否是JSON数据
	 * 如果是返回 JsonNode
	 * 如果不是返回null
	 * @param mapper
	 * @param message
	 * @return
	 */
	public static JsonNode isJson(ObjectMapper mapper,String message) {
		try {
			if(StringUtils.isEmpty(message)) {
				return null;
			}
			return mapper.readTree(message);
		} catch (Exception e) {
			LOG.warn("EMQ message type error: {}, message: {}",e.getMessage(),message);
			return null;
		}
	}

	public String getRoutingKey() {
		return routingKey;
	}

	public void setRoutingKey(String routingKey) {
		this.routingKey = routingKey;
	}
	
}
